package org.poem.zk;

import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import org.I0Itec.zkclient.ZkClient;
import org.poem.config.IZkLockConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

public class ZkParent {

    private static final Logger logger = LoggerFactory.getLogger(ZkParent.class);

    /**
     * 初始化锁
     */
    private Object initLock = new Object();
    /**
     * zk客户端集合
     */
    private static Map<String, ZkClient> zkClientMap = new HashMap<String, ZkClient>();
    /**
     * zk客户端关联实例
     */
    private static Map<ZkClient, Set<ZkParent>> zkReference = new HashMap<ZkClient, Set<ZkParent>>();

    /**
     * zk客户端
     */
    private ZkClient client;
    /**
     * zk连接配置
     */
    private IZkLockConfig config;

    public ZkParent(IZkLockConfig config) {
        validateConfig(config);
        if (!zkClientMap.containsKey(config.getZkServers())) {
            synchronized (initLock) {
                if (!zkClientMap.containsKey(config.getZkServers())) {
                    logger.debug("初始化zookeeper连接,zkservers:{},zkconnectTimeout:{}", config.getZkServers(), config.getZkConnectTimeout());
                    client = new ZkClient(config.getZkServers(),
                            config.getZkConnectTimeout() != null ? config.getZkConnectTimeout() : Integer.MAX_VALUE);
                    zkClientMap.put(config.getZkServers(), client);
                } else {
                    logger.debug("zookeeper连接已存在从池中获取,zkservers:{},zkconnectTimeout:{}", config.getZkServers(), config.getZkConnectTimeout());
                }
            }
        } else {
            logger.debug("zookeeper连接已存在从池中获取,zkservers:{},zkconnectTimeout:{}", config.getZkServers(), config.getZkConnectTimeout());
        }
        if (!zkReference.containsKey(client)) {
            synchronized (initLock) {
                if (!zkReference.containsKey(client)) {
                    zkReference.put(client, new HashSet<ZkParent>());
                }
            }
        }
        zkReference.get(client).add(this);
        client = zkClientMap.get(config.getZkServers());
        this.config = config;
    }

    /**
     * 检查配置是否可以使用
     *
     * @param config
     */
    private void validateConfig(IZkLockConfig config) {
        Preconditions.checkNotNull(config, "ZkDistributedLock 配置信息不能为空");
        Preconditions.checkArgument(!Strings.isNullOrEmpty(config.getZkServers()),
                "ZkDistributedLock 需指定zookeeper服务地址");
    }

    /**
     * 获取zookeeper 客户端
     *
     * @return ZkClient
     */
    protected ZkClient getZkClient() {
        return client;
    }

    /**
     * 获取zk连接配置
     *
     * @return IZkLockConfig
     */
    protected IZkLockConfig getConfig() {
        return config;
    }

    /**
     * 销毁对象，移除当前实例与zk客户端的关联，如zk客户端无其它关联则关闭并移除zk客户端;
     *
     * @return void
     */
    public void destroy() {
        logger.debug("zookeeper连接销毁,zkservers:{},zkconnectTimeout:{}", config.getZkServers(), config.getZkConnectTimeout());
        zkReference.get(client).remove(this);
        if (zkReference.get(client).isEmpty()) {
            synchronized (initLock) {
                if (zkReference.get(client).isEmpty()) {
                    zkClientMap.remove(client);
                    zkReference.remove(client);
                    client.close();
                }
            }
        }
    }
}
